package com.gandong8.trace.tool.config.kafka;

import com.gandong8.trace.tool.config.GlobalProperties;
import com.gandong8.trace.tool.config.kafka.properties.KafkaConsumerCommonData;
import com.gandong8.trace.tool.config.kafka.properties.KafkaConsumerProperties;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;

import java.util.HashMap;
import java.util.Map;

@EnableConfigurationProperties({KafkaConsumerProperties.class, GlobalProperties.class})
@Configuration
@Slf4j
public class KafkaConsumerConf {

    @Autowired
    KafkaConsumerProperties publicProp;

    @Autowired
    GlobalProperties globalProperties;

    private void dealConf(KafkaConsumerCommonData myProp) {
        if(myProp.getServers() == null) {
            myProp.setServers(publicProp.getServers());
        }
        if(myProp.getGroupId() == null) {
            myProp.setGroupId(publicProp.getGroupId());
        }
        if(myProp.getAutoCommit() == null) {
            myProp.setAutoCommit(publicProp.getAutoCommit());
        }
        if(myProp.getOffsetReset() == null) {
            myProp.setOffsetReset(publicProp.getOffsetReset());
        }
        if(myProp.getSessionTimeout() == null) {
            myProp.setSessionTimeout(publicProp.getSessionTimeout());
        }
        if(myProp.getAutoCommitInterval() == null) {
            myProp.setAutoCommitInterval(publicProp.getAutoCommitInterval());
        }
        if(myProp.getKeyDeserializer() == null) {
            myProp.setKeyDeserializer(publicProp.getKeyDeserializer());
        }
        if(myProp.getValueDeserializer() == null) {
            myProp.setValueDeserializer(publicProp.getValueDeserializer());
        }
        if(myProp.getAckMode() == null) {
            myProp.setAckMode(publicProp.getAckMode());
        }
    }



    private  Map<String, Object> consumerProps(KafkaConsumerCommonData myConf) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, myConf.getServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, myConf.getGroupId());
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, myConf.getClientId());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, myConf.getAutoCommit());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, myConf.getOffsetReset());
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, myConf.getSessionTimeout());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, myConf.getAutoCommitInterval());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, myConf.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, myConf.getValueDeserializer());
        return props;
    }

    /**
     *  RECORD:当每一条记录被消费者监听器处理之后提交
     *  BATCH:当每一批数据被消费者监听器处理之后提交
     *  TIME:当每一批数据被消费者监听器处理之后，距离上次提交时间大于TIME时提交
     *  COUNT:当每一批数据被消费者监听器处理之后，被处理record数量大于等于COUNT时提交
     *  COUNT_TIME：TIME 或 COUNT　有一个条件满足时提交
     *  MANUAL：当每一批数据被消费者监听器处理之后, 手动调用Acknowledgment.acknowledge()后提交
     *  MANUAL_IMMEDIATE:手动调用Acknowledgment.acknowledge()后立即提交
     * @param ackMode
     */
    private ContainerProperties.AckMode parseAckMode(String ackMode) {
        try {
            return ContainerProperties.AckMode.valueOf(ackMode);
        } catch (Exception e) {
            return ContainerProperties.AckMode.MANUAL_IMMEDIATE;
        }
    }


    public KafkaMessageListenerContainer createConsumer(KafkaConsumerCommonData myConf, MessageListener listener){
        if(!globalProperties.getKafkaConsumerOpen()) {
            System.err.println("kafka consumer not allowed");
            return null;
        }
        try {
            dealConf(myConf);
            ContainerProperties containerProps = new ContainerProperties(myConf.getTopics());
            containerProps.setMessageListener(listener);
            //若设置手动提交，忘记配置会出现Acknowledgment为null的情况
            containerProps.setAckMode(parseAckMode(myConf.getAckMode()));
            Map<String, Object> props = consumerProps(myConf);
            DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(props);
            KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(cf, containerProps);
            //container.setBeanName(myConf.getBeanName());
            container.start();
            log.info("----kafka consumer start with topic:{}----", myConf.getTopics());
            return container;
        } catch (Exception e) {
            log.error("kafka consumer create error", e);
            return null;
        }
    }





}
